#include <include/base-api/cf_base_types.h>
#include <devel-headers/cf_frame_types.h>
#include <devel-headers/cf_frame_api.h>
#include <pthread.h>
#include <string.h>


static const unsigned HASH_TABLE_MIN_SHIFT = 3; /* 1 << 3 == 8 buckets */

typedef struct simple_hash_node
{
    void *key;
	cf_item_handle_t *item;
	struct simple_hash_node *next;
} simple_hash_node_t;

typedef struct simple_hash_bucket
{
	simple_hash_node_t *head;
	pthread_spinlock_t lock;
} simple_hash_bucket_t;

typedef struct simple_hash_table
{
	unsigned long nnodes;
	unsigned long noccupied;
	simple_hash_bucket_t nodes[0];
} simple_hash_table_t;

static const unsigned prime_mod [] =
{
  1,          /* For 1 << 0 */
  2,
  3,
  7,
  13,
  31,
  61,
  127,
  251,
  509,
  1021,
  2039,
  4093,
  8191,
  16381,
  32749,
  65521,      /* For 1 << 16 */
  131071,
  262139,
  524287,
  1048573,
  2097143,
  4194301,
  8388593,
  16777213,
  33554393,
  67108859,
  134217689,
  268435399,
  536870909,
  1073741789,
  2147483647  /* For 1 << 31 */
};


static inline unsigned find_closest_shift(unsigned n)
{
  unsigned i;
  for (i = 0; n; i++)
    n >>= 1;
  return i;
}

static inline simple_hash_table_t *get_table(cacheflex_t *cache)
{
	simple_hash_table_t *table = (simple_hash_table_t *)cf_get_index_instance(cache); 
	if (!table)
	{
		CF_WRITE_LOG(CF_ERROR, "simple_hash::%s the index instance is NULL", __func__);

	}
	return table;
}



static inline simple_hash_bucket_t *get_bucket(cacheflex_t *cache, void *key)
{
	unsigned long node_index = 0;
	simple_hash_bucket_t *bucket;
    simple_hash_table_t *table;

    table = (simple_hash_table_t *)cf_get_index_instance(cache);
	node_index = CF_GET_KV_OPS(cache)->hash_key(key) % table->nnodes;
	bucket = &table->nodes[node_index];
	return bucket;
}



static int simple_hash_put(cacheflex_t *cache, void *key, cf_item_handle_t *ih)
{
	simple_hash_table_t *table;
	simple_hash_bucket_t *bucket;
	simple_hash_node_t *new_node = NULL;
	
	if (!ih)
	{
		CF_WRITE_LOG(CF_ERROR, "simple_hash::%s ih is NULL", __func__);
		return EINVAL;
	}
	table = get_table(cache);
	if (!table)
		return EINVAL;
	bucket = get_bucket(key, table);
	new_node = (simple_hash_node_t *)cf_malloc(sizeof(simple_hash_node_t));
	if (!new_node) {
		return ENOMEM;
	}
    
	/* memcpy(&new_node->item, ih, sizeof(item_handle_t)); */
    new_node->item = ih;
    new_node->key = key;

	pthread_spin_lock(&bucket->lock);
	if (bucket->head)
		new_node->next = bucket->head;
	bucket->head = new_node;
	pthread_spin_unlock(&bucket->lock);
	__sync_fetch_and_add(&table->noccupied, 1);
	return 0;
}

static int simple_hash_get(cacheflex_t *cache, void *key, cf_item_handle_t **ih)
{
	simple_hash_table_t *table;
	simple_hash_bucket_t *bucket;
	int ret = ENOENT;
	table = get_table(cache);
	if (!table)
		return EINVAL;
	bucket = get_bucket(key, table);

	pthread_spin_lock(&bucket->lock);
	simple_hash_node_t *node = bucket->head;
	while (node)
	{
		if (CF_GET_KV_OPS(cache)->compare_key(key, node->item.key))
			node = node->next;
		else
		{
			ret = 0;
			break;
		}
	}
	pthread_spin_unlock(&bucket->lock);
	if (ih && *ih && node)
        *ih = node->item; 
	return ret;
}

static int simple_hash_erase(cacheflex_t *cache, void *key)
{
	simple_hash_table_t *table;
	simple_hash_bucket_t *bucket;
	simple_hash_node_t *prev = NULL, *node = NULL;
	int ret = ENOENT;


	table = get_table(cache);
	if (!table)
		return EINVAL;
	bucket = get_bucket(key, table);

	pthread_spin_lock(&bucket->lock);
	node = prev = bucket->head;
	while (node)
	{
		if (CF_GET_KV_OPS(cache)->compare_key(key, node->item->key))
		{
			prev = node;
			node = node->next;
		}
		else
		{
			ret = 0;
			break;
		}
	}
	if (node && prev)
		prev->next = node->next;

	pthread_spin_unlock(&bucket->lock);
	if (node)
	{
        __sync_fetch_add_sub(&table->noccupied, 1);
        /*
        
		CF_STORE(free, cache, node->item->sh);
        */
		cf_free(node);
	}
	return ret;
}

static void simple_hash_destruct(void *instance)
{
    simple_hash_table_t *table = (simple_hash_table_t *)instance;
    simple_hash_node_t *node, *tmp;
    int i = 0;

    table->noccupied = 0;
    __sync_synchnorize();
    for (i = 0; i < table->nnodes; i++)
    {
        simple_hash_bucket_t *bucket = &table->nodes[i];
        pthread_spin_lock(&bucket->lock);
        tmp = node = bucket->head;
        while (node)
        {
            node = tmp->next;
            /*
             * CF_STORE(free, cache, tmp->item->sh);
             */
            cf_free(tmp);
            tmp = node;
        }
        bucket->head = NULL;
        pthread_spin_unlock(&bucket->lock);
    }
    table->nnodes = 0;
    return;
}

static int simple_hash_scan(cacheflex_t *cache, cf_index_scan_callback_pt scan_callback, void *args)
{
    int i = 0;
    simple_hash_table_t *table;
    simple_hash_node_t *node;
    simple_hash_bucket_t *bucket;
    table = get_table(cache);
    if (!table)
    {
        CF_WRITE_LOG(CF_ERROR, "simple_hash::%s cache is null", __func__);
        return EINVAL;
    }
    for (i = 0; i < table->nnodes; i++)
    {
        bucket = &table->nodes[i];
        pthread_lock(&bucket->lock);
        node = bucket->head;
        while (node)
        {
            scan_callback(node->item.key, node->item, args);
        }
        pthread_unlock(&bucket->lock);
    }
    return 0;
}


void *simple_hash_construct(const int64_t size, cf_index_interface_t *interface)
{
	simple_hash_table_t *table; 
	unsigned long nnodes = 0;
	unsigned long index = 0;
	
	nnodes = prime_mod[
		max(HASH_TABLE_MIN_SHIFT, find_closest_shift(size))];

	table = (simple_hash_table_t *)cf_malloc(sizeof(simple_hash_table_t) + nnodes * sizeof(simple_hash_bucket_t));
	if (!table) 
		return NULL;

	for (index = 0; index < nnodes; index++) {
		simple_hash_bucket_t *bucket = &table->nodes[index];
		pthread_spin_init(&bucket->lock, PTHREAD_PROCESS_SHARED);
		bucket->head = NULL;
	}
 
	table->nnodes = nnodes;
	table->noccupied = 0;

    interface->construct = simple_hash_construct;
    interface->put = simple_hash_put;
    interface->get = simple_hash_get;
    interface->erase = simple_hash_erase;
    interface->destruct = simple_hash_destruct;

	return (void *)table;
}
